package com.wchen.project.zk.curator;

import com.google.common.collect.Lists;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Desc:   使用 Curator 实现分布式 Master 选举
 *          注意：takeLeadership 执行完成后，Curator 就会立即释放 Master 权力，然后重新开始新的一轮的 Master 选举
 * @author: cww
 * @DateTime: 2020-01-24 22:16
 */

public class MasterSelectUseCurator {
    static String master_path = "/curator_recipes_master_path";
    static CuratorFramework client = CuratorFrameworkFactory.builder().connectString("home:2181")
            .retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();

    public static void main(String[] args) throws Exception {
        client.start();
        // LeaderSelectorListenerAdapter 在 Curator 成功获取到 Master 权利的时候回调该监听器
        LeaderSelector selector = new LeaderSelector(client, master_path, new LeaderSelectorListenerAdapter() {
            @Override
            public void takeLeadership(CuratorFramework client) throws Exception {
                System.out.println("成为 Master 角色");
//                GetConfigBuilder config = client.getConfig();
//                System.out.println("config:" + config.toString());

                Thread.sleep(2000);
                System.out.println("完成 Master 操作，释放 Master 权利");
            }
        });
        selector.autoRequeue();
        selector.start();
        Thread.sleep(4000);

        selector.autoRequeue();
        Thread.sleep(4000);
    }
}

class LeaderSelectorTest {
    static int CLIENT_COUNT = 10;
    static String LOCK_PATH = "/leader_selector";

    public static void main(String[] args) throws Exception{
        List<CuratorFramework> clientList = Lists.newArrayListWithCapacity(CLIENT_COUNT);
        // 启动 10 个 zk 客户端，每几秒进行一次 leader 选举
        for (int i = 0; i < CLIENT_COUNT; i++) {
            CuratorFramework client = getZkClient();
            clientList.add(client);
            ExampleClient exampleClient = new ExampleClient(client, LOCK_PATH, "client_" + i);
            System.out.println("add one client");
            exampleClient.start();
        }
        // sleep 以观察抢主过程
        Thread.sleep(10000);
    }

    static class ExampleClient extends LeaderSelectorListenerAdapter implements Cloneable {
        private final String name;
        private final LeaderSelector leaderSelector;
        private final AtomicInteger leaderCount = new AtomicInteger();

        public ExampleClient(CuratorFramework client,  String path, String name) {
            this.name = name;
            this.leaderSelector = new LeaderSelector(client, path, this);
            // 该方法能让客户端在释放 leader 权限后，重新加入 leader 权限的争夺中
            leaderSelector.autoRequeue();
        }

        public void start() throws IOException {
            leaderSelector.start();
        }

        public void  close() throws Exception {
            leaderSelector.close();
        }

        @Override
        public void takeLeadership(CuratorFramework client) throws Exception {
            // 抢到 leader 权限后 sleep 一段时间，并释放 leader 权限
            final int  waitSeconds = (int) ((5 * Math.random()) + 1);

            System.out.println(name + " is now the leader. Waiting " + waitSeconds + " seconds...");
            System.out.println(name + " has bean leader " + leaderCount.getAndIncrement() + " time(S) before.");
            try {
                Thread.sleep(TimeUnit.SECONDS.toMillis(waitSeconds));
            } catch (InterruptedException e) {
                System.err.println(name + " was interrupted");
                Thread.currentThread().interrupt();
            } finally {
                System.out.println(name + " relinquishing leadership.\n");
            }

        }

    }

    private static CuratorFramework getZkClient() {
        String zkServerAddress = "home:2181,home:2182,home:2183";
        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3, 5000);
        CuratorFramework zkClient = CuratorFrameworkFactory.builder()
                .connectString(zkServerAddress)
                .sessionTimeoutMs(5000)
                .connectionTimeoutMs(5000)
                .retryPolicy(retryPolicy)
                .build();
        zkClient.start();
        return zkClient;
    }

}




























